%%%-------------------------------------------------------------------
%%% File    : vect_ptt.erl
%%% Author  :  <vjache>
%%% Description : 
%%%
%%% Created : 24 Apr 2009 by  <vjache>
%%%-------------------------------------------------------------------
-module(vect_ptt).

%%--------------------------------------------------------------------
%% Include files
%%--------------------------------------------------------------------
-include("../include/vect.hrl").

%%--------------------------------------------------------------------
%% Call back API functions exports for 'vect' module 
%%--------------------------------------------------------------------
-export([
		 new/2,
		 new/4,
		 new/5,
		 close/1,
		 delete/2,
		 insert/2,
		 lookup/2,
		 delete_key/2,
		 all/1,
		 clear/1
		 ]).
%%--------------------------------------------------------------------
%% Test functions
%%--------------------------------------------------------------------
-export([test/0,test/1]).

%% -record(ptt_manager,{module,})

%%====================================================================
%% API
%%====================================================================
%%--------------------------------------------------------------------
%% Function: 
%% Description:
%%--------------------------------------------------------------------
new(VectName,Attrs) ->
	new(VectName,VectName,Attrs,undefined).

new(VectName,RecordName,Attrs,FragNum) ->
	new(VectName,RecordName,Attrs,FragNum,pmm_default).

new(VectName,RecordName,Attrs,FragProps,PMM) ->
	#vect{
		  drv_module=?MODULE,
		  open_fun=fun()->
						   _Session={
									 VectName,
									 PMM,
									 _PiVectSess=PMM:open_part_index(VectName,RecordName,Attrs,FragProps),
									 _PidOwner=make_ref()
									}
				   end,
		  close_fun=fun close/1,
		  delete_fun=fun delete/2,
		  delete_key_fun=fun delete_key/2,
		  insert_fun=fun insert/2,
		  lookup_fun=fun lookup/2,
		  all_fun=fun all/1,
		  clear_fun=fun clear/1
		  }.

cache_id(PiVectSess,_PidOwner) ->
	{PiVectSess}.


close({_VectName,PMM,PiVectSess,PidOwner}=_Session) ->
	ok=PMM:close_part_index(PiVectSess),
	ok=PMM:close_opened_partitions(PiVectSess,self()),
	ok=cache_ets:cleanup(_CacheId=cache_id(PiVectSess,PidOwner)).

delete({_VectName,PMM,PiVectSess,PidOwner}=_Session, Tuple) ->
	PartVectSess=get_partition(PMM,PiVectSess,PidOwner,Tuple),
	ok=vect:delete(PartVectSess,Tuple).

insert({_VectName,PMM,PiVectSess,PidOwner}=_Session, Tuples) when is_list(Tuples)->
%% 	io:format("DBG PTT INSERT ~p~p~n",[Tuples,node()]),
	PToolState=ptool:open(ptool_local:new(fun(T)->
									   {_PartVectSess=get_partition(PMM,PiVectSess,PidOwner,T),
										_Key=element(2,T)}
							   end)),
	ptool:pkeyfold(PToolState, 
				   _FoldFun=fun(_Key,Tuple,PartTuples) ->
						   [PMM:format(PiVectSess,Tuple)|PartTuples]
				   end, 
				   _FinishFun=fun({PartVectSess,_Key},FTuples) ->
						   ok=vect:insert(PartVectSess,FTuples)
				   end, [], Tuples),
	ok=ptool:close(PToolState);
insert({_VectName,PMM,PiVectSess,PidOwner}=_Session, Tuple) when is_tuple(Tuple)->
%% 	io:format("DBG PTT INSERT ~p ~p~n",[Tuple,node()]),
	PartVectSess=get_partition(PMM,PiVectSess,PidOwner,Tuple),
	ok=vect:insert(PartVectSess,PMM:format(PiVectSess,Tuple)).
	
delete_key({_VectName,PMM,PiVectSess,PidOwner}=_Session, Key) ->
	PartVectSess=get_partition_by_key(PMM,PiVectSess,PidOwner,Key),
	ok=vect:clear(PartVectSess).


all({_VectName,PMM,PiVectSess,_PidOwner}=_Session) -> 
	PartVectSessList=PMM:get_all_partitions(PiVectSess),
	iter_concat:new(lists:foldl(
					  fun(PartVectSess,ItersIn)->
							  [iter_mapper:new(vect:all(PartVectSess), 
											   fun(FTuple)->
													   PMM:deformat(PiVectSess,FTuple)
											   end)|ItersIn]
					  end, [], PartVectSessList)).

lookup({_VectName,PMM,PiVectSess,PidOwner}=_Session, Key) ->
	PartVectSess=get_partition_by_key(PMM,PiVectSess,PidOwner,Key),
	iter_mapper:new(vect:all(PartVectSess), 
					fun(FTuple)->
							PMM:deformat(PiVectSess,FTuple)
					end).

clear({_VectName,PMM,PiVectSess,_PidOwner}=_Session) ->
	PartVectSessList=PMM:get_all_partitions(PiVectSess),
	RpcRes=rpc:parallel_eval([{vect, clear, [PartVectSess]}||PartVectSess<-PartVectSessList]),
	case lists:all(fun(Res)->
					  Res==ok
			  end, RpcRes) of
		true ->
			ok;
		false ->
			exit({vect_clean_fail,RpcRes})
	end.

	
%%====================================================================
%% Internal functions
%%====================================================================

get_partition(PMM,PiVectSess,PidOwner,Tuple) ->
	get_partition_by_key(PMM,PiVectSess,PidOwner,element(2,Tuple)).
get_partition_by_key(PMM,PiVectSess,PidOwner,Key) ->
	cache_ets:obtain(_CacheId=cache_id(PiVectSess,PidOwner), 
					 Key, 
					 {PMM,get_partition_by_key,[PiVectSess,Key]}).

%%====================================================================
%% Test functions
%%====================================================================

test() ->
%% 	[Name,psog,[p,s,o,g],FragNum]
	V=new(predicate,predicate,[p,s,o,g],2),
	VSess=vect:open(V),
	ok=vect:insert(VSess,{predicate,"pred","subj","obj","graph"}),
	ok=vect:insert(VSess,{predicate,"pred","subj1","obj","graph"}),
	ok=vect:insert(VSess,{predicate,"pred","subj2","obj--","graph"}),
	ok=vect:insert(VSess,{predicate,"pred","subj3","obj74","graph"}),
	ok=vect:insert(VSess,{predicate,"pred","subj4","obj45","graph"}),
	ok=vect:insert(VSess,{predicate,"pred","subj5","obj%%","graph"}),
	ok=vect:insert(VSess,{predicate,"pred","subj6","obj","graph"}),
	ok=vect:insert(VSess,{predicate,"pred1","subj","obj","graph"}),
	ok=vect:insert(VSess,{predicate,"pred1","subj1","obj","graph"}),
	ok=vect:insert(VSess,{predicate,"pred1","subj2","obj--","graph"}),
	ok=vect:insert(VSess,{predicate,"pred2","subj3","obj74","graph"}),
	ok=vect:insert(VSess,{predicate,"pred2","subj4","obj45","graph"}),
	ok=vect:insert(VSess,{predicate,"pred2","subj5","obj%%","graph"}),
	ok=vect:insert(VSess,{predicate,"pred2","subj6","obj","graph"}),
	ok=vect:close(VSess).
test(list) ->
	V=new(predicate,predicate,[p,s,o,g],2),
	VSess=vect:open(V),
	List=[
	{predicate,"pred","subj","obj","graph"},
	{predicate,"pred","subj1","obj","graph"},
	{predicate,"pred","subj2","obj--","graph"},
	{predicate,"pred","subj3","obj74","graph"},
	{predicate,"pred","subj4","obj45","graph"},
	{predicate,"pred","subj5","obj%%","graph"},
	{predicate,"pred","subj6","obj","graph"},
	{predicate,"pred1","subj","obj","graph"},
	{predicate,"pred1","subj1","obj","graph"},
	{predicate,"pred1","subj2","obj--","graph"},
	{predicate,"pred2","subj3","obj74","graph"},
	{predicate,"pred2","subj4","obj45","graph"},
	{predicate,"pred2","subj5","obj%%","graph"},
	{predicate,"pred2","subj6","obj","graph"}],
	ok=vect:insert(VSess,List),
	ok=vect:close(VSess);

test(read_all) ->
	V=new(predicate,[p,s,o,g]),
	VSess=vect:open(V),
	iter:print(vect:all(VSess)),
	ok=vect:close(VSess);

test(read_key) ->
	V=new(predicate,[p,s,o,g]),
	VSess=vect:open(V),
	iter:print(vect:lookup(VSess,"pred1")),
	ok=vect:close(VSess);

test(del_key) ->
	V=new(predicate,[p,s,o,g]),
	VSess=vect:open(V),
	ok=vect:delete_key(VSess,"pred1"),
	iter:print(vect:all(VSess)),
	ok=vect:close(VSess);

test(clear) ->
	V=new(predicate,[p,s,o,g]),
	VSess=vect:open(V),
	ok=vect:clear(VSess),
	iter:print(vect:all(VSess)),
	ok=vect:close(VSess).

